use std::collections::HashMap;
use std::error::Error;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{Mutex, RwLock};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncReadExt, AsyncWriteExt, ReadHalf};
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::time;
use tracing::{error, info, warn};

pub mod config;
pub mod app;

#[derive(Serialize,Deserialize,Debug)]
enum DataType {
    PROVIDER,
    VISITOR
}

#[derive(Serialize,Deserialize,Debug)]
struct Data{
    data_type: DataType,
    name:String
}

impl Data{
    fn from_json(json:&str) -> serde_json::Result<Self> {
        serde_json::from_str(json)
    }

    fn to_json(&self) -> serde_json::Result<String>{
        serde_json::to_string(self)
    }
}



const HEARTBEAT_TIMEOUT: u64 = 60; // 心跳超时时间(秒)
async fn handle_provider(
    name: &str,
    mut stream: TcpStream,
    addr: SocketAddr,
    providers: ProviderMap,
)
{
    // 子层级，方便释放锁
    {
        let mut map = providers.lock().await;
        if map.contains_key(name){
            let rep = format!("name({})已经存在", name);
            info!("{rep}");
            if let Err(e) = stream.write_all(rep.as_bytes()).await{
                error!("消息发送失败{e}")
            }
            return;
        }
        map.insert(name.to_string(),addr);
    }

    // 心跳检测循环
    let mut last_active = time::Instant::now();
    loop {
        let mut buffer = [0u8; 4];
        let timeout = time::sleep_until(last_active + Duration::from_secs(HEARTBEAT_TIMEOUT));
        tokio::select! {
            result = stream.read(&mut buffer) => {
                match result {
                    Ok(0) => break,  // 连接关闭
                    Ok(n) => {
                        if &buffer[..n] != b"ping" {
                            error!("[{}] 无效心跳 heartbeat", addr);
                            break;
                        }
                        // info!("来心跳了:{}",addr);
                        last_active = time::Instant::now();
                    }
                    Err(e) => {
                        error!("[{}] Read error: {}", addr, e);
                        break;
                    }
                }
            }
            _ = timeout => {
                error!("[{}] Heartbeat timeout", addr);
                break;
            }
        }
    }

    // 清理资源
    providers.lock().await.remove(name);
    info!("[{}] Provider '{}' 已取消注册", addr, name);

}

async fn handle_visitor(
    name: &str,
    mut stream: TcpStream,
    providers: ProviderMap,
){

    let addr = match stream.peer_addr(){
        Ok(addr) => addr.to_string(),
        Err(e) => {
            error!("获取visitor地址错误：{}",e);
            return;
        }
    };

    let rep = match providers.lock().await.get(name) {
        None => String::from("NULL"),
        Some(addr) => addr.to_string()
    };

    if let Err(e) = stream.write_all(rep.as_bytes()).await{
        error!("handle_visitor {}",e);
    }

    // 对于visitor也需要保持连接
    // 心跳检测循环
    let mut last_active = time::Instant::now();
    loop {
        let mut buffer = [0u8; 4];
        let timeout = time::sleep_until(last_active + Duration::from_secs(HEARTBEAT_TIMEOUT));
        tokio::select! {
            result = stream.read(&mut buffer) => {
                match result {
                    Ok(0) => break,  // 连接关闭
                    Ok(n) => {
                        if &buffer[..n] != b"ping" {
                            error!("[{}] 无效心跳 heartbeat", addr);
                            break;
                        }
                        // info!("来心跳了:{}",addr);
                        last_active = time::Instant::now();
                    }
                    Err(e) => {
                        error!("[{}] Read error: {}", addr, e);
                        break;
                    }
                }
            }
            _ = timeout => {
                error!("[{}] Heartbeat timeout", addr);
                break;
            }
        }
    }

    warn!("visitr （{}）已端口连接",addr)
}

async fn get_auth(stream:&mut TcpStream) -> Result<Data,Box<dyn Error>>{
    let len = stream.read_i32().await?;
    let mut buf = vec![0;len as usize];
    stream.read(&mut buf).await?;
    let json = String::from_utf8_lossy(&buf);
    let data = Data::from_json(&json)?;
    Ok(data)
}

type ProviderMap = Arc<Mutex<HashMap<String,SocketAddr>>>;
pub async fn start(addr:&str) -> Result<(),Box<dyn Error>>{
    let listener = TcpListener::bind(addr).await?;
    info!("端口监听在 {addr}");
    let providers:ProviderMap = Arc::new(Mutex::new(HashMap::new()));
    loop {
        let providers_clone = Arc::clone(&providers);
        let (mut stream,addr) = listener.accept().await?;
        if let Ok(auth) = get_auth(&mut stream).await{
            info!("客户端{addr}信息 {:?}",auth);
            match auth.data_type {
                DataType::PROVIDER => {
                    tokio::spawn(async move{
                        handle_provider(&auth.name,stream,addr,providers_clone).await
                    });
                }
                DataType::VISITOR => {
                    tokio::spawn(async move{
                        handle_visitor(&auth.name,stream,providers_clone).await
                    });
                }
            }
        }

    }
}
